package control

import (
	"encoding/json"
	"git.oschina.net/ant180/scan/config"
	"github.com/nsqio/go-nsq"
	"github.com/xlvector/dlog"
	"net"
	"time"
)

type Reciver struct {
	conf *config.ReciverConf

	rfile       []*Message
	nsqconsumer *nsq.Consumer
	isRedis     bool
	isSSDB      bool
}

func (p *Reciver) Stop() {
	if p.nsqconsumer != nil {
		p.nsqconsumer.Stop()
	}
}

func (p *Worker) ReciverSSDB() {
	keys := p.reciver.conf.SSDB.Keys
	interval := p.reciver.conf.SSDB.Interval
	num := p.reciver.conf.SSDB.PopNum
	for {
		for _, k := range keys {
			for i := 0; i < num; i++ {
				val, err := p.model.SSDB.LPop(k)
				if err != nil || len(val) == 0 {
					dlog.Warn("ssdb pop fail! %v", err)
					break
				}

				m := &Message{}
				err = json.Unmarshal([]byte(val), m)
				if err != nil {
					dlog.Warn("ssdb pop unmarshal fail! %v %v", val, err)
					break
				}
				select {
				case p.msgch <- m:
				case <-p.exitch:
					dlog.Warn("reciver ssdb exit....")
					return
				}
			}
			time.Sleep(time.Duration(interval) * time.Second)
		}
	}
}
func (p *Worker) ReciverFile() {
	for {
		dlog.Info("reciver file begin......")
		startTime := time.Now().Unix()

		for _, msg := range p.reciver.rfile {
			select {
			case p.msgch <- msg:
			case <-p.exitch:
				dlog.Warn("reciverFile exit....")
				return
			}
		}

		s := p.reciver.conf.File.SumTime - (time.Now().Unix() - startTime)
		if s > 0 {
			time.Sleep(time.Duration(s) * time.Second)
		}
	}

}
func (p *Worker) ReciverRedis() {
	keys := p.reciver.conf.Redis.Keys
	interval := p.reciver.conf.Redis.Interval
	num := p.reciver.conf.Redis.PopNum
	for {
		for _, k := range keys {
			for i := 0; i < num; i++ {
				val, err := p.model.Redis.LPop(k)
				if err != nil {
					dlog.Warn("redis pop fail! %v", err)
					break
				}
				m := &Message{}
				err = json.Unmarshal([]byte(val), m)
				if err != nil {
					dlog.Warn("redis pop unmarshal fail! %v %v", val, err)
					break
				}

				select {
				case p.msgch <- m:
				case <-p.exitch:
					dlog.Warn("reciver redis exit....")
					return
				}
			}
			time.Sleep(time.Duration(interval) * time.Second)
		}
	}
}

func (p *Worker) ReciverStart() {
	if p.reciver.nsqconsumer != nil {
		p.reciver.nsqconsumer.AddHandler(p)
		err := p.reciver.nsqconsumer.ConnectToNSQD(p.reciver.conf.Nsq.Server)
		if err != nil {
			dlog.Fatal("connect nsq server fail! %v", err)
		}
	}

	if p.reciver.rfile != nil {
		go p.ReciverFile()
	}

	if p.reciver.isRedis && p.model.Redis != nil {
		go p.ReciverRedis()
	}

	if p.reciver.isSSDB && p.model.SSDB != nil {
		p.ReciverSSDB()
	}

}

func newNsqConsumer(nsqconf *config.ReciverNsqConf) (*nsq.Consumer, error) {
	config := nsq.NewConfig()
	config.LocalAddr, _ = net.ResolveTCPAddr("tcp", ":0")
	c, err := nsq.NewConsumer(nsqconf.Topic, nsqconf.Chan, config)
	if err != nil {
		dlog.Error("new nsq consumer fail! %v", err)
		return nil, err
	}

	return c, nil
}

func NewReciver(conf *config.ReciverConf) (*Reciver, error) {
	r := &Reciver{
		conf:        conf,
		rfile:       nil,
		nsqconsumer: nil,
		isSSDB:      false,
		isRedis:     false,
	}

	if len(conf.File.Name) > 0 {
		r.rfile = make([]*Message, 0, 1024)
		err := config.ReadConfig(conf.File.Name, &r.rfile)
		dlog.Info("reciver file: %v %v", r.rfile, err)
		if err != nil {
			return nil, err
		}
	}

	if len(conf.Nsq.Server) != 0 && len(conf.Nsq.Topic) != 0 && len(conf.Nsq.Chan) != 0 {
		c, err := newNsqConsumer(&conf.Nsq)
		if err != nil {
			return nil, err
		}
		r.nsqconsumer = c
	}

	if len(conf.SSDB.Keys) != 0 && conf.SSDB.PopNum != 0 {
		r.isSSDB = true
	}

	if len(conf.Redis.Keys) != 0 && conf.Redis.PopNum != 0 {
		r.isRedis = true
	}

	return r, nil
}

func (p *Worker) HandleMessage(msg *nsq.Message) error {
	m := &Message{}
	err := json.Unmarshal(msg.Body, m)
	if err != nil {
		dlog.Warn("nsq handler msg fail! %v %s", err, string(msg.Body))
		return nil
	}
	p.msgch <- m
	return nil
}
